package net.bwie.realtime.vehicle.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * @Author: FuHe
 * @Date: 2025/5/27
 */
public class VehicleLogClickHouse {
    public static void sinkToClickhouse(DataStream<String> etlStream) {
        SingleOutputStreamOperator<String> vehicleStream = etlStream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        JSONObject vehicleObject = JSON.parseObject(value);
                        Double longitude = vehicleObject.getDouble("longitude");
                        Double latitude = vehicleObject.getDouble("latitude");

                        JSONObject jsonObject = new JSONObject();

                        jsonObject.put("vin", vehicleObject.getString("vin"));
                        jsonObject.put("ts", vehicleObject.getLong("timestamp"));
                        jsonObject.put("longitude", longitude);
                        jsonObject.put("latitude", latitude);
                        jsonObject.put("speed", vehicleObject.getDouble("speed"));
                        jsonObject.put("mileage", vehicleObject.getDouble("mileage"));
                        jsonObject.put("batteryLevel", vehicleObject.getDouble("batteryLevel"));
                        jsonObject.put("batteryTemp", vehicleObject.getDouble("batteryTemp"));
                        jsonObject.put("motorTemp", vehicleObject.getDouble("motorTemp"));
                        jsonObject.put("chargingStatus", vehicleObject.getString("chargingStatus"));
                        jsonObject.put("energyConsumption", vehicleObject.getDouble("energyConsumption"));
                        jsonObject.put("startTime", vehicleObject.getString("startTime"));
                        jsonObject.put("endTime", vehicleObject.getString("endTime"));
                        jsonObject.put("batteryHealth", vehicleObject.getDouble("batteryHealth"));

                        VehicleLogEtlJob.SubstringFunction substringFunction = new VehicleLogEtlJob.SubstringFunction();
                        String geohash = substringFunction.evaluate(longitude, latitude, 6);
                        jsonObject.put("geohash", geohash);

                        return jsonObject.toJSONString();
                    }
                }
        );
        // 字段名称
        String[] columns = {
                "vin", "ts", "longitude", "latitude", "speed", "mileage", "batteryLevel", "batteryTemp",
                "motorTemp", "chargingStatus", "energyConsumption", "startTime", "endTime",
                "batteryHealth", "geohash"
        };
        // 插入语句
        String upserSql = "INSERT INTO vehicle_log.vehicle_date_log (" +
                "vin, ts, longitude, latitude, speed, mileage, batteryLevel, batteryTemp,\n" +
                "motorTemp, chargingStatus, energyConsumption, startTime, endTime,\n" +
                "batteryHealth, geohash)\n" +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);";
        JdbcUtil.sinkToClickhouseUpsertJson(vehicleStream, columns, upserSql);
    }
}
